#-*- coding:utf-8 -*-
# author: wucheng@staff.sina.com.cn
'''最近开始大数据处理工作,从数据库6,7千万的数据中筛选需要的数据,所以写了个python多进程程序用来处理这些问题. 同时把这个简陋的程序分享出来,要是大家只希望能有个能干活的程序的话,重写一下producer和worker函数即可.'''
import multiprocessing
import os
import sys
import Queue
import time
def writeQ(q,obj):
	q.put(obj,True,None)
	print "put size: ",q.qsize()
def readQ(q):
	ret = q.get(True,1)
	print "get size: ",q.qsize()
	return ret
def producer(q):
	'''生产者函数，将资源从famous_uid.txt取出来，然后丢到q队列里面去。'''
	time.sleep(5)   #让进行休息几秒 方便ps命令看到相关内容
	pid = os.getpid()
	handle_file = 'famous_uid.txt'
	with open(handle_file,'r') as f:
		for line in f:
			print "producer <" ,pid , "> is doing: ",line
			writeQ(q,line.strip())
	q.close()
def worker(q):
	'''消费者函数，将资源从q队列里面取出来处理'''
	time.sleep(5)   #让进行休息几秒 方便ps命令看到相关内容	
	pid = os.getpid()
	empty_count = 0
	while True:
		try:
			task = readQ(q)
			print "worker <" , pid , "> is doing: " ,task
			'''如果这里不休眠的话 一般情况下所有行都会被同一个子进程读取到为了使实验效果更加清楚 在这里让每个进程读取完一行内容时候休眠5s 这样就可以让其他的进程到队列中进行读取'''
			time.sleep(5)
		except Queue.Empty:
			empty_count += 1
			if empty_count == 3:
				print "queue is empty, quit"
				q.close()
				sys.exit(0)
def main():
	concurrence = 3
	q = multiprocessing.Queue(10)
	funcs = [producer , worker]
	for i in range(concurrence-1):
		funcs.append(worker)
	for item in funcs:
		print str(item)
	nfuncs = range( len(funcs) )
	processes = [] 
	for i in nfuncs:
		p = multiprocessing.Process(target=funcs[i] , args=(q,))
		processes.append(p)
	print "concurrence worker is : ",concurrence," working start"
	for i in nfuncs:
		processes[i].start()
	for i in nfuncs:
		processes[i].join()
	print "all DONE" 
if __name__ == '__main__':
	main()	
